package javaVersion.learn.stateProcess.operatorStateDemo;

import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.runtime.state.StateBackend;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @program: myFlink
 * @description: operatorState实现offset偏移量管理
 * @author: WincoMa
 * @create: 2020-07-27 15:20
 **/
public class StreamOperatorStateDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment senv = StreamExecutionEnvironment.getExecutionEnvironment();
        senv.enableCheckpointing(5000);
        senv.setParallelism(1);
        //设置状态后端
        StateBackend fsStateBackend = new FsStateBackend("file:///D:/WorkSpace/myFlink//data");
        senv.setStateBackend(fsStateBackend);
        //设置ck
        CheckpointConfig ck = senv.getCheckpointConfig();
        ck.enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        //重启策略
        senv.setRestartStrategy(RestartStrategies.fixedDelayRestart(3,5000));

        DataStreamSource<String> source = senv.addSource(new MyRichSource());
        source.print();
        senv.execute();


    }
}
